﻿//
// Copyright (c) 2024 BPer
//

#include <bperice/BPerRtcSocket.h>

#include <bperutil/sys/BPerLog.h>
#include <bperutil/sys/BPerEndian.h>
#include <bperrtp/BPerRtpConstant.h>

#include <errno.h>



int32_t bPer_rtc_sendData(BPerRtcSocketSession *session, char *data, int32_t nb) {
	if (session == NULL || !session->isStart || session->notRemoteInit || data==NULL)	return ERROR_RTC_SOCKET;
	int32_t err = BPer_Ok;
	bPer_thread_mutex_lock(&session->sendLock);
	err=bPer_socket_sendto(session->fd, data, nb, &session->remote_addr,0) > 0 ? BPer_Ok : ERROR_RTC_SOCKET;
	bPer_thread_mutex_unlock(&session->sendLock);
	return err;
}


int32_t bPer_rtc_tcp_sendData(BPerRtcSocketSession *session,char *data, int32_t nb) {
	if (session == NULL || !session->isStart || session->notRemoteInit || data==NULL)	return ERROR_RTC_SOCKET;
	bPer_thread_mutex_lock(&session->sendLock);
	int32_t nbytes=bPer_socket_send(session->fd, data, nb);
	bPer_thread_mutex_unlock(&session->sendLock);
	if(nbytes==-1) bPer_error("rtc tcp send error(%d)",nbytes);
	if(nbytes!=nb) bPer_error("rtc tcp send data(%d),should send data(%d)",nbytes,nb);
	return nbytes==nb ? BPer_Ok : ERROR_RTC_SOCKET;
}

#if BPer_Enable_Tcp_Srs
	int32_t bPer_rtc_tcp_sendData_srs(BPerRtcSocketSession *session,char *data, int32_t nb) {
		if (session == NULL || !session->isStart || session->notRemoteInit || data==NULL || session->fd==-1)	return ERROR_RTC_SOCKET;

		bPer_thread_mutex_lock(&session->sendLock);
		int32_t sendBytes=2;

		bPer_put_be16(session->buffer,(uint16_t)nb);

		int32_t nbytes=bPer_socket_send(session->fd, session->buffer, sendBytes);

		if(nbytes==-1 ||nbytes!=sendBytes) goto fail;

		sendBytes = nb;
		nbytes=bPer_socket_send(session->fd, data, sendBytes);// BPer_No_Signal) ;

		if(nbytes==-1 ||nbytes!=sendBytes) goto fail;

		bPer_thread_mutex_unlock(&session->sendLock);
		return BPer_Ok;

		fail:
			bPer_thread_mutex_unlock(&session->sendLock);
			if(nbytes==-1) return bPer_error_wrap(ERROR_RTC_SOCKET,"rtc tcp send error(%d)",nbytes);
			if(nbytes!=sendBytes) return bPer_error_wrap(ERROR_RTC_SOCKET,"rtc tcp send data(%d),should send data(%d)",nbytes,sendBytes);
			return BPer_Ok;



	}
#endif

void* bPer_run_rtcudp_thread(void *obj) {
    BPerRtcSocketSession *sock = (BPerRtcSocketSession*) obj;

	sock->isStart = BPerTrue;

	if (bPer_socket_listen(sock->fd, &sock->local_addr) < 0) {

		bPer_error("Udp server bind error");
		exit(1);
	}

	char buffer[2048] = { 0 };
	sock->isLoop = BPerTrue;

	int32_t len = 0;
	if (sock->startStunTimer)	sock->startStunTimer(sock->user);

	bPer_trace("\nrtc udp(%s) server is starting,localPort=%d",sock->familyType==BPer_IpFamilyType_IPV4?"IPV4":"IPV6", sock->local_addr.port);
	BPerIpAddress src;
	bPer_memset(&src, 0, sizeof(BPerIpAddress));
	src.familyType=sock->familyType;
	src.protocol=BPer_Socket_Protocol_Udp;
	while (sock->isLoop) {
		bPer_memset(buffer, 0, 2048);
		if ((len = bPer_socket_recvfrom(sock->fd, buffer, 2048,  &src)) > 0) {

			if(sock->notRemoteInit){
				bPer_memcpy((char*)&sock->remote_addr,(char*)&src,sizeof(BPerIpAddress));
				sock->notRemoteInit=BPerFalse;
			}
			if (sock->receive)	sock->receive(buffer, len, sock->user);
		}
	}

	sock->isStart = BPerFalse;
	bPer_socket_close(sock->fd);
	sock->fd = -1;

	return NULL;
}


void* bPer_run_rtctcp_thread(void *obj) {

    BPerRtcSocketSession *sock = (BPerRtcSocketSession*) obj;
    char buffer[4096] = { 0 };
	sock->isStart = BPerTrue;

	if (bPer_socket_connect(sock->fd, &sock->remote_addr) == -1) {
		bPer_socket_close(sock->fd);
		bPer_error_wrap(ERROR_SOCKET, "rtc tcp connect socket error(%d)",GetSockError());
		return NULL;
	}

	if (!sock->notRemoteInit&&sock->startStunTimer)	sock->startStunTimer(sock->user);

	bPer_trace("\n rtc tcp socket(%s) is starting",sock->familyType==BPer_IpFamilyType_IPV4?"IPV4":"IPV6");
	sock->isLoop = BPerTrue;
	int32_t readLen = 0;

	while (sock->isLoop) {
		bPer_memset(buffer, 0, 4096);
        if ((readLen = bPer_socket_recv(sock->fd, buffer, 4096, 0)) > 0) {
        		if (sock->receive)	sock->receive(buffer, readLen, sock->user);

		}else if (readLen == -1) {
			int32_t sockerr = GetSockError();
			if (sockerr == EINTR)
				continue;
			if (sockerr == EWOULDBLOCK || sockerr == EAGAIN) {
				readLen = 0;
				continue;
			}
			bPer_error("%s, recv returned %d. GetSockError(): %d (%s)",
					__FUNCTION__, readLen, sockerr, strerror(sockerr));

			break;
		}
	}

	sock->isStart = BPerFalse;
	bPer_socket_close(sock->fd);
	sock->fd = -1;

	return NULL;
}



void* bPer_run_rtctcp_thread_srs(void *obj) {

    BPerRtcSocketSession *sock = (BPerRtcSocketSession*) obj;
    char readBuffer[kRtpPacketSize] = { 0 };

	sock->isStart = BPerTrue;

	if (bPer_socket_connect(sock->fd, &sock->remote_addr) == -1) {
		bPer_socket_close(sock->fd);
		bPer_error_wrap(ERROR_SOCKET, "rtc tcp connect socket error(%d)",GetSockError());
		return NULL;
	}

	if (!sock->notRemoteInit&&sock->startStunTimer)	sock->startStunTimer(sock->user);

	bPer_trace("\n rtc tcp socket(%s) is starting",sock->familyType==BPer_IpFamilyType_IPV4?"IPV4":"IPV6");
	sock->isLoop = BPerTrue;
	int32_t readLen = 0;
	uint16_t msgLen=0;
	int32_t msgReadLen=0;
	int32_t msgReadBufferLen=0;
	BPerBool recvData=BPerFalse;
	int32_t headerLen=sock->headerLen;

	while (sock->isLoop) {
		if(!recvData){
			if(msgLen==0)
				msgReadLen=headerLen;
			else
				msgReadLen=msgLen;
		}
		if ((readLen = bPer_socket_recv(sock->fd, readBuffer+msgReadBufferLen, msgReadLen, 0))> 0) {
			if(readLen!=msgReadLen){
				msgReadLen-=readLen;
				recvData=BPerTrue;
				msgReadBufferLen+=readLen;
				continue;
			}
			msgReadBufferLen=0;
			recvData=BPerFalse;
			if (msgLen == 0) {
				msgLen = bPer_get_be16((uint8_t*)(readBuffer));
				if (msgLen > kRtpPacketSize) {
					bPer_error("rtc tcp error pkt size=%hu", msgLen);
					msgLen = 0;
					msgReadLen = 0;
				}
				continue;
			}

			if (sock->receive)
				sock->receive(readBuffer, msgLen, sock->user);

			msgLen = 0;
		}else if (readLen == -1) {
			int32_t sockerr = GetSockError();
			if (sockerr == EINTR)
				continue;
			if (sockerr == EWOULDBLOCK || sockerr == EAGAIN) {
				readLen = 0;
				continue;
			}
			bPer_error("%s, recv returned %d. GetSockError(): %d (%s)",
					__FUNCTION__, readLen, sockerr, strerror(sockerr));

			break;
		}
	}

	sock->isStart = BPerFalse;
	bPer_socket_close(sock->fd);
	sock->fd = -1;

	return NULL;
}

void* bPer_run_rtctcp_thread_turn(void *obj) {

    BPerRtcSocketSession *sock = (BPerRtcSocketSession*) obj;
    char readBuffer[kRtpPacketSize] = { 0 };

	sock->isStart = BPerTrue;

	if (bPer_socket_connect(sock->fd, &sock->remote_addr) == -1) {
		bPer_socket_close(sock->fd);
		bPer_error_wrap(ERROR_SOCKET, "rtc tcp connect socket error(%d)",GetSockError());
		return NULL;
	}

	if (!sock->notRemoteInit&&sock->startStunTimer)	sock->startStunTimer(sock->user);

	bPer_trace("\n rtc tcp socket(%s) is starting",sock->familyType==BPer_IpFamilyType_IPV4?"IPV4":"IPV6");
	sock->isLoop = BPerTrue;
	int32_t readLen = 0;
	int32_t msgLen=0;
	int32_t msgReadLen=0;
	int32_t msgReadBufferLen=0;
	BPerBool recvData=BPerFalse;
	int32_t headerLen=sock->headerLen;
	char* p=readBuffer;

	int32_t blank=0;
	while (sock->isLoop) {
		if(!recvData){
			if(msgLen==0)
				msgReadLen=headerLen;
			else
				msgReadLen=msgLen;
		}
		if ((readLen = bPer_socket_recv(sock->fd, p+msgReadBufferLen, msgReadLen, 0))> 0) {
			if(readLen!=msgReadLen){
				msgReadLen-=readLen;
				recvData=BPerTrue;
				msgReadBufferLen+=readLen;
				continue;
			}
			msgReadBufferLen=0;
			recvData=BPerFalse;
			if (msgLen == 0) {
				if(bPer_get_be32((uint8_t*) readBuffer + 4) == 0x2112A442){
					msgLen = bPer_get_be16((uint8_t*)readBuffer+2)+12;
					blank=0;

				}else{
					msgLen = (int32_t)bPer_get_be16((uint8_t*)readBuffer+2);

					msgLen-=4;
					blank=4-msgLen%4;

					if(blank<4) msgLen+=blank;
				}


				if (msgLen > kRtpPacketSize) {
					bPer_error("rtc tcp error pkt size=%d", msgLen);
					msgLen = 0;
					msgReadLen = 0;
					p=readBuffer;
					continue;
				}
				p=readBuffer+headerLen;
				continue;
			}

			if (sock->receive)
				sock->receive(readBuffer, msgLen+headerLen-blank, sock->user);
			p=readBuffer;
			msgLen = 0;
		}else if (readLen == -1) {
			int32_t sockerr = GetSockError();
			if (sockerr == EINTR)
				continue;
			if (sockerr == EWOULDBLOCK || sockerr == EAGAIN) {
				readLen = 0;
				continue;
			}
			bPer_error("%s, recv returned %d. GetSockError(): %d (%s)",
					__FUNCTION__, readLen, sockerr, strerror(sockerr));

			break;
		}
	}

	sock->isStart = BPerFalse;
	bPer_socket_close(sock->fd);
	sock->fd = -1;

	return NULL;
}

void bPer_start_rtcudp(BPerRtcSocketSession *sock) {
	if (sock == NULL || sock->isStart)	return;
	if(sock->socketProtocol==BPer_Socket_Protocol_Udp){
		if (bPer_thread_create(&sock->threadId, 0, bPer_run_rtcudp_thread, sock)) {
			bPer_error("BPerThread::start could not start thread");
		}
	}else if(sock->socketProtocol==BPer_Socket_Protocol_Tcp){

		if (bPer_thread_create(&sock->threadId, 0,
				sock->headerLen==2?bPer_run_rtctcp_thread_srs:bPer_run_rtctcp_thread_turn, sock))

		{
					bPer_error("BPerThread::start could not start thread");
		}


	}
}


void bPer_stop_rtcudp(BPerRtcSocketSession *sock) {
	if (sock == NULL)	return;
	sock->isLoop = BPerFalse;
	while (sock->isStart)	bPer_usleep(1000);
}

void bPer_rtcudp_update_remoteAddr(BPerRtcSocketSession *sock,char* remoteIp,int32_t port){

	bPer_trace("\nremoteIp=%s,port=%d",remoteIp,port);
	bPer_addr_set(&sock->remote_addr,remoteIp,port,sock->familyType,sock->socketProtocol);

	sock->notRemoteInit=(port==0?BPerTrue:BPerFalse);

}


int32_t bPer_create_rtcsocket(BPerRtcSocket *psock,BPerIpFamilyType familyType, BPerSocketProtocol protocol,int32_t plocalPort) {
	if (psock == NULL)		return ERROR_RTC_SOCKET;
	BPerRtcSocketSession* sock=&psock->session;

	sock->fd = -1;
	sock->socketProtocol=protocol;
	sock->familyType=familyType;

	sock->headerLen=0;

	bPer_addr_setAnyAddr(&sock->local_addr,plocalPort,familyType,protocol);

	sock->fd = bPer_socket_create(familyType,protocol);
	if(sock->fd==-1){
		return bPer_error_wrap(ERROR_RTC_SOCKET,"create rtc socket fail,familtype=%s,protocol=%s",
				familyType==BPer_IpFamilyType_IPV4?"IPV4":"IPV6",protocol==BPer_Socket_Protocol_Udp?"UDP":"TCP");
	}


	sock->notRemoteInit=BPerTrue;

	bPer_thread_mutex_init(&sock->sendLock,NULL);

	psock->start=bPer_start_rtcudp;
	psock->stop=bPer_stop_rtcudp;
	if(protocol == BPer_Socket_Protocol_Udp)
		psock->write=bPer_rtc_sendData;
	else
		psock->write=bPer_rtc_tcp_sendData;


	psock->updateRemoteAddress=bPer_rtcudp_update_remoteAddr;

	return BPer_Ok;

}

#if BPer_Enable_Tcp_Srs

int32_t bPer_create_rtcsocket_srs(BPerRtcSocket* psock,BPerSocketProtocol protocol){
	if(protocol!=BPer_Socket_Protocol_Tcp) return BPer_Ok;
	psock->session.headerLen=2;
	psock->write=bPer_rtc_tcp_sendData_srs;

	if(psock->session.buffer==NULL) psock->session.buffer=(char*)bPer_calloc(2,1);
	return BPer_Ok;
}

#endif

void bPer_destroy_rtcsocket(BPerRtcSocket *sock) {
    if (sock == NULL)        return;
    bPer_stop_rtcudp(&sock->session);
    if (sock->session.fd > 0) {
    	bPer_socket_close(sock->session.fd);
        sock->session.fd = -1;
    }
    bPer_free(sock->session.buffer);
    bPer_thread_mutex_destroy(&sock->session.sendLock);

}
